package com.pupu.kafka;

import com.pupu.kafka.entity.CoinGetEntity;
import com.pupu.kafka.entity.DiscountGetEntity;
import com.pupu.utils.JsonUtils;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.function.Function;

/**
 * @author chenzuyi
 */
@SuppressWarnings("ALL")
public class ProducerFastStart {

    public static final String bootstrapServers = "10.120.156.239:9092";
    public static final Integer retries = 1;
    public static final Integer batchSize = 16384;
    public static final Long lingerMs = 1L;
    public static final Long bufferMemory = 33554432L;
    public static final Class keySerializeClass = StringSerializer.class;
    public static final Class valueSerializeClass = StringSerializer.class;
    public static final String topic = "user_assets_event";
    public static final String DISCOUNT_ENTITY_GET = "DISCOUNT_ENTITY_GET";
    public static final String COIN_GET = "COIN_GET";

    private static final String discountId = "";
    private static final String coinId = "";

    /**
     * 启动入口
     */
    public static void main(String[] args) throws InterruptedException {
        KafkaTemplate<String, String> kafkaTemplate = kafkaTemplate();
        // 发送优惠卷
        discount(kafkaTemplate);

        // 发送积分
        // coin(kafkaTemplate);

        // 睡眠10秒关闭
        Thread.sleep(10000L);
    }

    private static void discount(KafkaTemplate<String, String> kafkaTemplate) {
        System.out.println("start...");
        long now = System.currentTimeMillis();
        for (int i = 0; i < 1; i++) {
            DiscountGetEntity discount = generateDiscount();
            Message<String> message = buildMessage(discount, DISCOUNT_ENTITY_GET, DiscountGetEntity::getUserId);
            kafkaTemplate.send(message);
        }
        System.out.println("end..." + (System.currentTimeMillis() - now) + "ms");
    }

    private static void coin(KafkaTemplate<String, String> kafkaTemplate) {
        System.out.println("start...");
        long now = System.currentTimeMillis();
        for (int i = 0; i < 1; i++) {
            CoinGetEntity coin = generateCoin();
            Message<String> message = buildMessage(coin, COIN_GET, CoinGetEntity::getUserId);
            kafkaTemplate.send(message);
        }
        System.out.println("end..." + (System.currentTimeMillis() - now) + "ms");
    }

    private static ProducerFactory<String, String> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializeClass);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializeClass);
        return new DefaultKafkaProducerFactory<>(props);
    }

    private static KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    private static <T> Message<String> buildMessage(T data, String msgType, Function<T, String> keyFunction) {
        return MessageBuilder.withPayload(JsonUtils.toJson(data))
                .setHeader(KafkaHeaders.MESSAGE_KEY, keyFunction.apply(data))
                .setHeader(KafkaHeaders.TOPIC, topic)
                .setHeader("type", msgType)
                .build();
    }

    private static CoinGetEntity generateCoin() {
        CoinGetEntity entity = new CoinGetEntity();
        entity.setId(coinId);
        entity.setUserId(UUID.randomUUID().toString());
        return entity;
    }

    private static DiscountGetEntity generateDiscount() {
        DiscountGetEntity entity = new DiscountGetEntity();
        entity.setId(discountId);
        entity.setUserId(UUID.randomUUID().toString());
        return entity;
    }


}
